package com.bff.gaia.streaming.api.xjoin.client.hbase.async.rowkeydealer;

import com.bff.gaia.api.common.typeinfo.TypeInformation;
import com.bff.gaia.api.java.tuple.Tuple3;
import com.bff.gaia.streaming.api.functions.async.ResultFuture;
import com.bff.gaia.streaming.api.xjoin.core.enums.EJoinType;
import com.bff.gaia.streaming.api.xjoin.core.side.FieldInfo;
import com.bff.gaia.streaming.api.xjoin.core.side.SideInfo;
import com.bff.gaia.streaming.api.xjoin.core.side.cache.AbsSideCache;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Maps;
import com.bff.gaia.types.Row;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public abstract class AbsRowKeyModeDealer {

	protected Map<String, String> colRefType;

	protected String[] colNames;

	protected HTable hBaseClient;

	protected boolean openCache;

	protected EJoinType joinType;

	protected List<FieldInfo> outFieldInfoList;

	protected int[] outFieldIndexs;

	protected SideInfo sideInfo;

	//key:Returns the value of the position, returns the index values ​​in the input data
	protected Map<Integer, Integer> inFieldIndex = Maps.newHashMap();

	protected Map<Integer, Integer> sideFieldIndex = Maps.newHashMap();

	public AbsRowKeyModeDealer(Map<String, String> colRefType, String[] colNames, HTable hBaseClient,
	                           boolean openCache,  SideInfo sideInfo){
		this.colRefType = colRefType;
		this.colNames = colNames;
		this.hBaseClient = hBaseClient;
		this.openCache = openCache;
		this.joinType = sideInfo.getJoinType();
		this.outFieldInfoList =  sideInfo.getOutFieldInfoList();
		this.inFieldIndex = sideInfo.getInFieldIndex();
		this.sideFieldIndex = sideInfo.getSideFieldIndex();
		this.outFieldIndexs=sideInfo.getOutFieldIndexs();
		this.sideInfo=sideInfo;
	}

	protected void dealMissKey(Row input, ResultFuture<Row> resultFuture){
		if(joinType == EJoinType.LEFT){
			//保留left 表数据
			Row row = fillData(input, null);
			resultFuture.complete(Collections.singleton(row));
		}else{
			resultFuture.complete(null);
		}
	}

	protected Row fillData(Row input, Object sideInput){

		List<Object> sideInputList = (List<Object>) sideInput;
		Row row = new Row(outFieldInfoList.size());
		for(Map.Entry<Integer, Integer> entry : inFieldIndex.entrySet()){
			Object obj = input.getField(entry.getValue());
			if(obj instanceof Timestamp){
				obj = ((Timestamp)obj).getTime();
			}
			row.setField(entry.getKey(), obj);
		}

		for(Map.Entry<Integer, Integer> entry : sideFieldIndex.entrySet()){
			if(sideInputList == null){
				row.setField(entry.getKey(), null);
			}else{
				row.setField(entry.getKey(), sideInputList.get(entry.getValue()));
			}
		}

		return row;
	}

	protected Get createGet(String rowKeyStr)  {
		// currently this rowKey always be a binary type (adapt to sql-level type system)
		Get get = new Get(rowKeyStr.getBytes());
		//TODO add request colums
		return get;
	}

	protected Map<String, Object> getResultMap(Result result){
		Map<String, Object> kv=new HashMap<>();
		for (Cell cell : result.listCells())
		{
			String family = Bytes.toString(CellUtil.cloneFamily(cell));
			String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));
			StringBuilder key = new StringBuilder();
			key.append(family).append(":").append(qualifier);
			kv.put(key.toString(), value);
		}
		return kv;
	}

	public abstract void asyncGetData(String tableName, String rowKeyStr, Row input, ResultFuture<Row> resultFuture,
	                                  AbsSideCache sideCache);
}